package com.atguigu.gmall.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.JdbcUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/9/29 10:20
 * 通过jdbc来实现
 */
public class PhoenixSink_Batch extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private Connection conn;
    private ValueState<Boolean> isFirstState;
    private int count;
    private ValueState<PreparedStatement> psState;
    
    //
    @Override
    public void open(Configuration parameters) throws Exception {
        // 创建Phoenix连接
        // 1. 加载Phoenix驱动
        
        // 2. 获取连接对象
        String url = Constant.PHOENIX_URL;
        conn = JdbcUtil.getPhoenixConnection(url);
        
        isFirstState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("isFirstState", Boolean.class));
    
        psState = getRuntimeContext().getState(new ValueStateDescriptor<PreparedStatement>("psState", PreparedStatement.class));
    
        count = 0;
        
    }
    
    // 流中每来一条数据, 执行一次
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value,
                       Context ctx) throws Exception {
        // 1. 检测要写的表是否存在, 不存在就需要先建表
        checkTable(value);
        // 2. 写入数据
        writeToPhoenix(value);
        
    }
    
    private void writeToPhoenix(Tuple2<JSONObject, TableProcess> value) throws SQLException, IOException {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
    
        if (psState.value() == null) {
            // 1. 拼接sql   "id,name"
            // upsert into t(id, name) values(?,?)
            StringBuilder sql = new StringBuilder();
            sql
                .append("upsert into ")
                .append(tp.getSink_table())
                .append("(")
                // 字段
                .append(tp.getSink_columns())
                .append(")values(")
                .append(tp.getSink_columns().replaceAll("[^,]+", "?"))
                .append(")");
    
            // 2. 获取PrepareStatement
            PreparedStatement ps = conn.prepareStatement(sql.toString());
    
            psState.update(ps);
        }
    
        PreparedStatement ps = psState.value();
        // 3. 给sql中的占位符赋值
        String[] columnNames = tp.getSink_columns().split(",");
        for (int i = 0; i < columnNames.length; i++) {
            Object v = data.get(columnNames[i]); // v有可能是null,
            // 1 -> "1"    null -> "null"
            ps.setString(i + 1, v == null ? null : v.toString());
        }
        // 4. 执行
        ps.addBatch();
        if(++count % 10 == 0){
            ps.executeBatch();
            conn.commit();
        }
        ps.close();
        
    }
    
    private void checkTable(Tuple2<JSONObject, TableProcess> value) throws SQLException, IOException {
        
        if (isFirstState.value() == null) {
            System.out.println("xxxx");
            // 建表只需要第一条数据过来的时候执行一次
            TableProcess tp = value.f1;
            // 主要工作建表
            // create table if not exists t(id varchar, name varchar, constraint pk primary key(id, name)) SALT_BUCKETS = 4;
            StringBuilder sql = new StringBuilder();
            sql
                .append("create table if not exists ")
                .append(tp.getSink_table())
                .append("(")
                .append(tp.getSink_columns().replaceAll(",", " varchar,"))
                .append(" varchar, constraint pk primary key(")
                .append(tp.getSink_pk() == null ? "id" : tp.getSink_pk())
                .append("))")
                .append(tp.getSink_extend() == null ? "" : tp.getSink_extend());
            
            PreparedStatement ps = conn.prepareStatement(sql.toString());
            ps.execute();
            conn.commit();
            ps.close();
            
            isFirstState.update(true);
        }
        
    }
    
    @Override
    public void close() throws Exception {
        if (conn != null) {
            conn.close();
        }
    }
}
